package com.yun.util.clickHouse;
import com.yun.util.ClassUtil;
import com.yun.util.clickHouse.annotation.CkColum;
import com.yun.util.clickHouse.annotation.CkTable;
import com.yun.util.clickHouse.bean.DbFiled;
import com.yun.util.clickHouse.bean.TableClassBean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.persistence.MappedSuperclass;
import javax.sql.DataSource;
import java.lang.reflect.Field;
import java.math.BigInteger;
import java.sql.*;
import java.util.Date;
import java.util.*;


@Component
public class CreateClickHouseTableConfig {

    @Autowired
    @Qualifier("clickhouseDataSource")
    private DataSource dataSource;
    private Connection connection ;

    @Value("${clickhouse.config.database}")
    private String database;
    @Value("${clickhouse.config.entityPackage}")
    private String entityPackage;

    static Logger log = LoggerFactory.getLogger(CreateClickHouseTableConfig.class);

    @PostConstruct
    public void initClickHouseTable() throws SQLException {
            connection = dataSource.getConnection();
            DatabaseMetaData metaData = connection.getMetaData();
            ResultSet tables = metaData.getTables(null, database, null, null);
            Set<String> tableSet = new HashSet<>();
            while (tables.next()) {
                String tableName = tables.getString("TABLE_NAME");
                tableSet.add(tableName);
            }
            Statement statement = connection.createStatement();
                List<Class<?>> classListByAnnotation = ClassUtil.getClassListByAnnotation(entityPackage, CkTable.class);
                for (Class<?> clazz :classListByAnnotation) {
                    TableClassBean tableClassBean = getTableClassBean(clazz);
                    if (tableSet.contains(tableClassBean.getTbName())){
                        //更新表
                        ResultSet columns = metaData.getColumns(null, database, tableClassBean.getTbName(), "%");
                        Set<String> columnList = new HashSet<>();
                        while (columns.next()) {
                            columnList.add(columns.getString("COLUMN_NAME"));

                        }
                        List<DbFiled> newDbFileds = new ArrayList<>();
                        for (DbFiled dbFiled:tableClassBean.getDbFiledList()) {
                            if (!columnList.contains(dbFiled.getColumName())){
                                newDbFileds.add(dbFiled);
                            }
                        }
                        if (newDbFileds.size()>0){
                            tableClassBean.setDbFiledList(newDbFileds);
                            String sql = CreatTableSqlUtil.getAlterTable(tableClassBean);
                            log.info("ClickHouse SQL: {}",sql);
                            statement.execute(sql);
                        }
                        if (tableClassBean.getTtlExprs().length>0){
                            String sql = CreatTableSqlUtil.getAlterTableTTL(tableClassBean);
                            log.info("ClickHouse SQL: {}",sql);
                            statement.execute(sql);
                        }

                    }else {
                        //新增表
                        String sql = CreatTableSqlUtil.getTableNameList(tableClassBean);
                        log.info("ClickHouse SQL: {}",sql);
                        statement.execute(sql);
                    }
                }
//            }
            statement.close();
            connection.close();
//        }
    }

    /**
     * 获取table中的columns和类中的字段集合(包括表名前提model配置@ClickHouseTable注解且注解包含name属性)
     *
     * @param clazz
     * @return
     */
    public TableClassBean getTableClassBean(Class<?> clazz) {
        List<DbFiled> list = new ArrayList<>();
        Field[] fields = clazz.getDeclaredFields();
        for (int j = 0; j < fields.length; j++) {
            Field field = fields[j];
            if (field.isAnnotationPresent(CkColum.class)) {
                CkColum annotation = field.getAnnotation(CkColum.class);
                list.add(new DbFiled(annotation.name(),annotation.columType().equals("") ? getCkColumType(field.getType()): annotation.columType(),annotation.nullAble()));

            }
        }
        Class<?> superClass = clazz.getSuperclass();
        while (superClass != Object.class ){
            if (!superClass.isAnnotationPresent(MappedSuperclass.class))
                continue;
            Field[] superFields = superClass.getDeclaredFields();
            for (Field superField : superFields) {
                if (superField.isAnnotationPresent(CkColum.class)) {
                    CkColum annotation = superField.getAnnotation(CkColum.class);
                    list.add(new DbFiled(annotation.name(),annotation.columType().equals("") ? getCkColumType(superField.getType()): annotation.columType(),annotation.nullAble()));
                }
            }
            superClass = superClass.getSuperclass();
        }


        CkTable annotationsByType = (CkTable) clazz.getAnnotation(CkTable.class);

            if (annotationsByType.name() != null && !"".equals(annotationsByType.name().trim())) {
                return new TableClassBean(database,annotationsByType.name(),annotationsByType.engine(),annotationsByType.parititionBy(),annotationsByType.orderBy(),annotationsByType.primaryBy(),annotationsByType.sample(),annotationsByType.settings(),annotationsByType.ttlExpr() ,list);
            }

        log.info("no annotation or javax.persistence.Table");
        return new TableClassBean(clazz.getSimpleName(), list);
    }


    public String getCkColumType(Class<?> type){
        if (boolean.class.equals(type) || Boolean.class.equals(type)) {
            return "UInt8";
        }else if (byte.class.equals(type) || Byte.class.equals(type)){
            return "Int8";
        }else if (short.class.equals(type) || Short.class.equals(type)){
            return "Int16";
        }else if (int.class.equals(type) || Integer.class.equals(type)){
            return "Int32";
        }else if (long.class.equals(type) || Long.class.equals(type) || BigInteger.class.equals(type)){
            return "Int64";
        }else if (float.class.equals(type) || Float.class.equals(type)){
            return "";

        }else if (double.class.equals(type) || Double.class.equals(type)){
            return "Float32";
        }else if (String.class.equals(type)){
            return "String";
        }else if (Date.class.equals(type)){
            return "Date";
        }else if (Timestamp.class.equals(type)){
            return "Datetime";
        }
        return "";
    }

}
